﻿package mqttclient

import (
	"errors"
	"fmt"
	"time"

	mqtt "github.com/eclipse/paho.mqtt.golang"
)

// MQTT客户端连接的相关参数信息
type MqttConnectionConfig struct {
	Broker           string
	User             string
	Password         string
	ClientID         string
	WillEnable       bool
	WillTopic        string
	WillPayload      string
	WillQos          byte
	Qos              byte
	Retained         bool
	OnConnect        mqtt.OnConnectHandler
	OnConnectionLost mqtt.ConnectionLostHandler
}

type MqttClient struct {
	qos      byte
	retained bool
	Client   mqtt.Client
	topics   map[string]mqtt.MessageHandler
}

func (mc *MqttClient) connectHandler(handler mqtt.OnConnectHandler) mqtt.OnConnectHandler {
	return func(c mqtt.Client) {
		for topic, onMessage := range mc.topics {
			mc.Client.Subscribe(topic, mc.qos, onMessage)
		}
		handler(c)
	}
}

func (mc *MqttClient) connectionLostHandler(handler mqtt.ConnectionLostHandler) mqtt.ConnectionLostHandler {
	return func(c mqtt.Client, err error) {
		handler(c, err)
	}
}

func (mc *MqttClient) Close() {
	mc.Client.Disconnect(200)
}

func (mc *MqttClient) Subscribe(topics []string, onMessage mqtt.MessageHandler) error {
	for _, topic := range topics {
		token := mc.Client.Subscribe(topic, mc.qos, onMessage)
		if token.Wait() && nil != token.Error() {
			return token.Error()
		}
		mc.topics[topic] = onMessage
	}
	return nil
}

func (mc *MqttClient) UnSubscribe(topics ...string) error {
	token := mc.Client.Unsubscribe(topics...)
	if token.Wait() && nil != token.Error() {
		return token.Error()
	}
	for _, topic := range topics {
		delete(mc.topics, topic)
	}
	return nil
}

func (mc *MqttClient) Publish(topic string, payload []byte) error {
	if nil != mc && mc.Client.IsConnected() {
		token := mc.Client.Publish(topic, mc.qos, mc.retained, payload)
		if token.Wait() && nil != token.Error() {
			fmt.Println("MQTT Publish Error")
			return token.Error()
		}
		return nil
	}
	return errors.New("Mqtt Client is nil or disconnected")
}

func NewMqttClient(config MqttConnectionConfig) (*MqttClient, error) {
	var client MqttClient
	opts := mqtt.NewClientOptions()
	opts.AddBroker(config.Broker)
	opts.SetClientID(config.ClientID)
	opts.SetMaxReconnectInterval(5 * time.Second)

	if config.WillEnable {
		opts.SetWill(config.WillTopic,
			config.WillPayload,
			config.WillQos,
			config.Retained)
	}

	if nil == config.OnConnect {
		config.OnConnect = func(c mqtt.Client) {}
	}

	if nil == config.OnConnectionLost {
		config.OnConnectionLost = func(c mqtt.Client, err error) {}
	}

	opts.SetOnConnectHandler(client.connectHandler(config.OnConnect))
	opts.SetConnectionLostHandler(client.connectionLostHandler(config.OnConnectionLost))
	client.Client = mqtt.NewClient(opts)
	client.qos = config.Qos
	client.retained = config.Retained
	client.topics = make(map[string]mqtt.MessageHandler)

	token := client.Client.Connect()
	if token.Wait() && nil != token.Error() {
		return nil, token.Error()
	}
	return &client, nil
}
